package com.tianji.aigc.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.tianji.aigc.config.SystemPromptConfig;
import com.tianji.aigc.contans.Constant;
import com.tianji.aigc.domain.vo.ChatEventVO;
import com.tianji.aigc.enums.ChatEventTypeEnum;
import com.tianji.aigc.service.IChatService;
import com.tianji.aigc.service.IChatSessionService;
import com.tianji.aigc.utils.ToolResultHolder;
import com.tianji.common.exceptions.RequestTimeoutException;
import com.tianji.common.utils.CollUtils;
import com.tianji.common.utils.UserContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.client.advisor.vectorstore.QuestionAnswerAdvisor;
import org.springframework.ai.chat.memory.ChatMemory;
import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.ai.vectorstore.SearchRequest;
import org.springframework.ai.vectorstore.VectorStore;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 读取配置项
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class ChatServiceImpl implements IChatService {

    private final ChatClient dashScopeChatClient;
    private final ChatClient ollamaChatClient;
    private final ChatClient openAiChatClient;
    private final SystemPromptConfig systemPromptConfig;
    private final ChatMemory chatMemory;
    private final VectorStore vectorStore;
    private final IChatSessionService chatSessionService;

    // 存储大模型的生成状态，这里采用ConcurrentHashMap是确保线程安全
    // 目前的版本暂时用Map实现，如果考虑分布式环境的话，可以考虑用redis来实现
    private static final Map<String, Boolean> GENERATE_STATUS = new ConcurrentHashMap<>();

    // 输出结束的标记
    private static final ChatEventVO STOP_EVENT = ChatEventVO.builder().eventType(ChatEventTypeEnum.STOP.getValue()).build();


    @Override
    public Flux<ChatEventVO> chat(String question, String sessionId) {
        //获取对话id
        String conversationId = IChatService.getConversationId(sessionId);
        // 大模型输出内容的缓存器，用于在输出中断后的数据存储
        var outputBuilder = new StringBuilder();
        // 生成请求id
        var requestId = IdUtil.fastSimpleUUID();
        // 获取用户id
        var userId = UserContext.getUser();
        chatSessionService.update(sessionId,question,userId);

        // 定义RAG查询增强
        var qaAdvisor = QuestionAnswerAdvisor.builder(vectorStore)
                .searchRequest(SearchRequest.builder()
                        .similarityThreshold(0.6d) // 设置相似度阈值
                        .topK(6)// 设置返回的文档数量
                        .build()).build();

        return dashScopeChatClient.prompt()
                .system(prompt -> prompt
                        .text(systemPromptConfig.getChatSystemMessage().get())// 设置系统提示语
                        .param("now", DateUtil.now()))// 设置当前时间
                .advisors(advisor -> advisor
                        .advisors(qaAdvisor) // 添加RAG增强
                        .param(ChatMemory.CONVERSATION_ID,conversationId))
                .toolContext(MapUtil.<String,Object>builder() // 设置tool列表
                        .put(Constant.REQUEST_ID,requestId) // 设置请求id参数
                        .put(Constant.USER_ID, UserContext.getUser()) // 设置用户id参数
                        .build()
                )
                .user(question)
                .stream()
                .chatResponse()
                .doFirst(()->{GENERATE_STATUS.put(sessionId, true);})// 输出开始,标记正在输出
                .doOnComplete(() -> {
                    Optional.ofNullable(sessionId).ifPresent(GENERATE_STATUS::remove);
                }) //输出结束,清除标记
                .doOnError(throwable -> {
                    Optional.ofNullable(sessionId).ifPresent(GENERATE_STATUS::remove);
                    log.error("用户会话异常:{}", sessionId, throwable);
                }) // 输出异常,清除标记
                .doOnCancel(()->{
                    // 当输出被取消时,保存输出的内容到历史记录中
                    saveStopHistoryRecord(conversationId,outputBuilder.toString());
                })
                // 输出过程中,判断是否正在输出,如果正在输出,则继续输出,否则结束输出
                .takeWhile(s -> Optional.ofNullable(GENERATE_STATUS.get(sessionId)).orElse(false))
                .map(chatResponse -> {
                    //对于乡音结果进行处理,如果是最后一条数据,就把此次消息id放到内存中
                    //主要用于存储消息数据到redis中,可以根据消息id获取请求id,再通过请求id就可以获取参数列表了
                    //从而解决,在历史聊天记录中没有外参数的问题
                    var finishReason = chatResponse.getResult().getMetadata().getFinishReason();
                    if (StrUtil.equals(Constant.STOP,finishReason)){
                        var messageId = chatResponse.getMetadata().getId();
                        ToolResultHolder.put(messageId, Constant.REQUEST_ID, requestId);
                    }
                    //获取大模型的输出内容
                    String text = chatResponse.getResult().getOutput().getText();
                    // 追加到输出内容中
                    outputBuilder.append(text);
                    // 封装成VO
                    return ChatEventVO.builder()
                            .eventData(text)
                            .eventType(ChatEventTypeEnum.DATA.getValue())
                            .build();
                })
                .concatWith(
                    //Flux.defer 确保了在流的最后阶段才会去检查和处理工具执行结果，而不是在流初始化时就执行
                    Flux.defer(()->{
                    //通过请求id获取到参数列表,如果列表不为空,就将其追加到返回结果中
                    var map = ToolResultHolder.get(requestId);
                    if (CollUtils.isNotEmpty(map)){
                        ToolResultHolder.remove(requestId);// 清除参数列表
                        // 响应给前端的参数数据
                        ChatEventVO chatEventVO = ChatEventVO.builder()
                                .eventData(map)
                                .eventType(ChatEventTypeEnum.PARAM.getValue())
                                .build();
                        return Flux.just(chatEventVO,STOP_EVENT);
                    }
                    return Flux.just(STOP_EVENT);
                }));
    }


    /**
     * 保存停止输出的记录
     *
     * @param conversationId 会话id
     * @param content        大模型输出的内容
     */
    private void saveStopHistoryRecord(String conversationId, String content) {
        this.chatMemory.add(conversationId, new AssistantMessage(content));
    }

    @Override
    public void stopChat(String sessionId) {
        // 移除标记
        GENERATE_STATUS.remove(sessionId);
    }

    @Override
    public String textChat(String question) {
        // 检查输入
        if (StrUtil.isBlank(question)) {
            throw new RequestTimeoutException(500,"请输入问题");
        }
        try {
            return this.openAiChatClient.prompt()
                    .system(promptSystemSpec ->
                            promptSystemSpec.text(systemPromptConfig.getTextSystemMessage().get()))
                    .user(question)
                    .call()
                    .content();
        } catch (Exception e) {
            log.error("文本聊天异常:{}", question, e);
            throw new RequestTimeoutException(500,"文本聊天异常");
        }
    }
}
